package mr.flow;

import java.io.IOException;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;

public class FlowCount {
	static class FlowCountMapper extends Mapper<LongWritable, Text, FlowBean, Text> {
		FlowBean k = new FlowBean();
		Text v = new Text();
		@Override
		protected void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			String line = value.toString();
			String[] fields = line.split(",");
			String phoneNbr = fields[0];
			k.set(Long.parseLong(fields[1]), Long.parseLong(fields[2]));
			v.set(phoneNbr);
			context.write(k, v);
		}
	}
	
	static class FlowCountReducer extends Reducer<FlowBean, Text, Text, FlowBean> {
		@Override
		protected void reduce(FlowBean bean, Iterable<Text> phoneNbr, Context context)
				throws IOException, InterruptedException {
			context.write(phoneNbr.iterator().next(), bean);
		}
	}
	
}
